#include "stdafx.h"
#include "net/TCPServerDispatcher.hpp"
#include "msg/notification.hpp"
using namespace JHCPP::msg;

NET_NAMESPACE_BEGIN


//
// CTCPConnectionNotification
//
class CTCPConnectionNotification : public CNotification
{
public:
	CTCPConnectionNotification(const CStreamSocket& socket) : m_socket(socket)
	{
	}

	~CTCPConnectionNotification()
	{
	}

	const CStreamSocket& socket() const
	{
		return m_socket;
	}

private:
	CStreamSocket m_socket;
};


//
// CTCPServerDispatcher
//
CTCPServerDispatcher::CTCPServerDispatcher(CTCPServerConnectionFactory::Ptr pFactory, CThreadPool& threadPool, CTCPServerParams::Ptr pParams)
	: m_rc(1),
	m_pParams(pParams),
	m_currentThreads(0),
	m_totalConnections(0),
	m_currentConnections(0),
	m_maxConcurrentConnections(0),
	m_refusedConnections(0),
	m_stopped(false),
	m_pConnectionFactory(pFactory),
	m_threadPool(threadPool)
{
	jh_check_ptr (pFactory);

	if (!m_pParams)
		m_pParams = new CTCPServerParams;

	if (m_pParams->getMaxThreads() == 0)
		m_pParams->setMaxThreads(threadPool.capacity());
}

CTCPServerDispatcher::~CTCPServerDispatcher()
{
}

void CTCPServerDispatcher::add()
{
	m_mutex.lock();
	++m_rc;
	m_mutex.unlock();
}

void CTCPServerDispatcher::release()
{
	m_mutex.lock();
	int rc = --m_rc;
	m_mutex.unlock();
	if (rc == 0) 
		delete this;
}

void CTCPServerDispatcher::run()
{
	TAutoPtr<CTCPServerDispatcher> guard(this, true); // ensure object stays alive

	int idleTime = (int) m_pParams->getThreadIdleTime().totalMilliseconds();

	for (;;)
	{
		TAutoPtr<CNotification> pNf = m_queue.waitDequeueNotification(idleTime);
		if (pNf)
		{
			CTCPConnectionNotification* pCNf = dynamic_cast<CTCPConnectionNotification*>(pNf.get());
			if (pCNf)
			{
				std::auto_ptr<CTCPServerConnection> pConnection(m_pConnectionFactory->createConnection(pCNf->socket()));
				jh_check_ptr(pConnection.get());
				beginConnection();
				pConnection->start();
				endConnection();
			}
		}

		CMutex::ScopedLock lock(m_mutex);
		if (m_stopped || (m_currentThreads > 1 && m_queue.empty()))
		{
			--m_currentThreads;
			break;
		}
	}
}

static const std::string threadName("TCPServerConnection");
void CTCPServerDispatcher::enqueue(const CStreamSocket& socket)
{
	CMutex::ScopedLock lock(m_mutex);

	if (m_queue.size() < m_pParams->getMaxQueued())
	{
		m_queue.enqueueNotification(new CTCPConnectionNotification(socket));
		if (!m_queue.hasIdleThreads() && m_currentThreads < m_pParams->getMaxThreads())
		{
			try
			{
				m_threadPool.startWithPriority(m_pParams->getThreadPriority(), *this, threadName);
				++m_currentThreads;
			}
			catch (CException&)
			{
				// no problem here, connection is already queued
				// and a new thread might be available later.
			}
		}
	}
	else
	{
		++m_refusedConnections;
	}
}

void CTCPServerDispatcher::stop()
{
	m_stopped = true;
	m_queue.clear();
	m_queue.wakeUpAll();
}

int CTCPServerDispatcher::currentThreads() const
{
	CMutex::ScopedLock lock(m_mutex);

	return m_currentThreads;
}

int CTCPServerDispatcher::maxThreads() const
{
	CMutex::ScopedLock lock(m_mutex);

	return m_threadPool.capacity();
}

int CTCPServerDispatcher::totalConnections() const
{
	CMutex::ScopedLock lock(m_mutex);

	return m_totalConnections;
}

int CTCPServerDispatcher::currentConnections() const
{
	CMutex::ScopedLock lock(m_mutex);

	return m_currentConnections;
}

int CTCPServerDispatcher::maxConcurrentConnections() const
{
	CMutex::ScopedLock lock(m_mutex);

	return m_maxConcurrentConnections;
}

int CTCPServerDispatcher::queuedConnections() const
{
	return m_queue.size();
}

int CTCPServerDispatcher::refusedConnections() const
{
	CMutex::ScopedLock lock(m_mutex);

	return m_refusedConnections;
}

const CTCPServerParams& CTCPServerDispatcher::params() const
{
	return *m_pParams;
}

void CTCPServerDispatcher::beginConnection()
{
	CMutex::ScopedLock lock(m_mutex);

	++m_totalConnections;
	++m_currentConnections;
	if (m_currentConnections > m_maxConcurrentConnections)
		m_maxConcurrentConnections = m_currentConnections;
}

void CTCPServerDispatcher::endConnection()
{
	CMutex::ScopedLock lock(m_mutex);

	--m_currentConnections;
}

NET_NAMESPACE_END
